package websocket

import (
	"bytes"
	"compress/flate"
	"context"
	"encoding/json"
	"net/http"
	"sync"
	"time"

	"github.com/gogf/gf/v2/container/gset"
	"github.com/gogf/gf/v2/encoding/gjson"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/net/ghttp"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/gogf/gf/v2/text/gstr"
	"github.com/gogf/gf/v2/util/gconv"
	"github.com/gorilla/websocket"

	"github.com/beherochuling/beesoft_api/app/helper/helper"
)

type handleOnOpenFunc func(ws *Websocket)
type handleOnCloseFunc func(ws *Websocket)
type handleOnErrorFunc func(ws *Websocket, err error)
type handleOnMessageFunc func(ws *Websocket, data *ActionMessage)
type handleOnGroupOpenFunc func(ws *Websocket, groupId string)
type handleOnGroupCloseFunc func(closedGroupId string, allClosed bool)
type handleOnGroupJoinFunc func(ws *Websocket, data *ActionMessage, joinedGroupId string)
type handleOnGroupExitFunc func(ws *Websocket, data *ActionMessage, exitedGroupId string)

func OnOpen(handle handleOnOpenFunc) {
	callbackOnOpen = handle
}
func OnClose(handle handleOnCloseFunc) {
	callbackOnClose = handle
}
func OnError(handle handleOnErrorFunc) {
	callbackOnError = handle
}
func OnMessage(handle handleOnMessageFunc) {
	callbackOnMessage = handle
}
func OnGroupOpen(handle handleOnGroupOpenFunc) {
	callbackOnGroupOpen = handle
}
func OnGroupClose(handle handleOnGroupCloseFunc) {
	callbackOnGroupClose = handle
}
func OnGroupJoin(handle handleOnGroupJoinFunc) {
	callbackOnGroupJoin = handle
}
func OnGroupExit(handle handleOnGroupExitFunc) {
	callbackOnGroupExit = handle
}

var (
	callbackOnError      handleOnErrorFunc
	callbackOnOpen       handleOnOpenFunc
	callbackOnClose      handleOnCloseFunc
	callbackOnMessage    handleOnMessageFunc
	callbackOnGroupJoin  handleOnGroupJoinFunc
	callbackOnGroupExit  handleOnGroupExitFunc
	callbackOnGroupOpen  handleOnGroupOpenFunc
	callbackOnGroupClose handleOnGroupCloseFunc

	ctx             = context.Background()
	once            = &sync.Once{}
	pingPongBytes   = []byte("p")
	sonyIdStartTime = gtime.ParseTimeFromContent("2022-11-18", "Y-m-d").Time

	heartbeatTimeout    = g.Config().MustGet(ctx, "app.websocket.heartbeatTimeout", 30).Uint()
	enableContentFilter = g.Config().MustGet(ctx, "app.websocket.enableContentFilter", false).Bool()
	enableCompression   = g.Config().MustGet(ctx, "app.websocket.enableCompression", true).Bool()
)

func Init(s *ghttp.Server) {
	once.Do(func() {
		enabled := g.Config().MustGet(ctx, "app.websocket.enabled", false).Bool()
		if !enabled {
			return
		}

		pattern := g.Config().MustGet(ctx, "app.websocket.router", uriPrefix).String()
		s.BindHandler(pattern, func(r *ghttp.Request) {
			wsUpgrader := websocket.Upgrader{
				EnableCompression: enableCompression,
				CheckOrigin: func(r *http.Request) bool {
					return true
				},
			}
			conn, err := wsUpgrader.Upgrade(r.Response.Writer, r.Request, nil)
			if err != nil {
				r.Response.WriteStatus(200, err.Error())
				return
			}

			ws := &ghttp.WebSocket{
				Conn: conn,
			}
			if enableCompression {
				_ = ws.SetCompressionLevel(flate.BestCompression)
			}

			es := &Websocket{
				mux:       &sync.Mutex{},
				FD:        int64(helper.Faker.GetSonyID(sonyIdStartTime)), // atomicClientFD.Add(1), //uuid.New().String(), //string(getSocketFD(ws.UnderlyingConn())),
				Request:   r,
				websocket: ws,
				UserId:    gconv.Int(r.Context().Value("user_id")),
				GroupId:   r.Get("group_id").String(),
				groupIds:  gset.NewStrSet(true),
			}

			if !es.groupIds.Contains(es.GroupId) {
				es.groupIds.Add(es.GroupId)

				if callbackOnGroupOpen != nil { // ☆ 房间创建事件
					callbackOnGroupOpen(es, es.GroupId)
				}
			}

			onOpened(es)
			for {
				msgType, msg, err := ws.ReadMessage()
				if err != nil {
					onErrorOccurred(es, err)
					return
				} else if msgType == websocket.TextMessage {
					onMessageReceived(es, msg) // 处理消息
				}
			}
		})

		if cluster.Enabled {
			cluster.Run(s)
		}
	})
}

func onOpened(es *Websocket) {
	if es.GroupId != "" {
		group.Add(es.GroupId, es)
	}
	if callbackOnOpen != nil { // ☆ 打开事件
		callbackOnOpen(es)
	}

	keepAlive(es)
}
func onClosed(es *Websocket) {
	if es.GroupId != "" {
		group.Remove(es.GroupId, es.FD)
	}
	if callbackOnClose != nil {
		callbackOnClose(es)
	}
	es.close()

	if callbackOnGroupClose != nil { // ☆ 房间关闭事件
		exists := group.GroupExists(messageFromLocal, es.GroupId)
		callbackOnGroupClose(es.GroupId, !exists)
	}
}
func onErrorOccurred(es *Websocket, err error) {
	if websocket.IsCloseError(err) ||
		websocket.IsUnexpectedCloseError(err) ||
		es.closed ||
		gstr.ContainsI(err.Error(), "use of closed network connection") || // 当心跳超时被关闭后，有可能触发该错误
		gstr.ContainsI(err.Error(), "operation timed out") { // 在无心跳的情况下，客户端网络不通，有可能触发该错误
		onClosed(es) // ☆ 关闭事件
	} else if callbackOnError != nil {
		callbackOnError(es, err) // ☆ 错误事件
	}
}
func onMessageReceived(es *Websocket, msg []byte) {
	res := bytes.Compare(msg, pingPongBytes)
	if res == 0 {
		es.callbackOnReceivedKeepAlive(true)
		return
	} else {
		es.callbackOnReceivedKeepAlive(false)
	}

	if callbackOnMessage != nil {
		var actionMessage *ActionMessage
		err := gconv.Struct(msg, &actionMessage)
		if err == nil && actionMessage.Action != 0 {
			if actionMessage.Action == ActionJoinRoom || actionMessage.Action == ActionLeaveRoom {
				j := gjson.New(actionMessage.Data)
				groupId := gstr.Trim(j.Get("group_id").String())
				if groupId != "" {
					if actionMessage.Action == ActionJoinRoom {
						es.GroupId = groupId
						group.Add(groupId, es)
						es.groupIds.AddIfNotExist(groupId)
						if callbackOnGroupJoin != nil { // ☆ 房间加入事件
							callbackOnGroupJoin(es, actionMessage, groupId)
						}
					} else { // 从房间退出
						group.Remove(groupId, es.FD)
						es.groupIds.Remove(groupId)
						if es.groupIds.Size() > 0 {
							es.GroupId = es.groupIds.Slice()[0]
						} else {
							es.GroupId = ""
						}
						if callbackOnGroupExit != nil { // ☆ 房间退出事件
							callbackOnGroupExit(es, actionMessage, groupId)
						}
					}
				}

				return
			}

			if actionMessage.Options != nil {
				if len(actionMessage.Options.To) > 0 {
					for _, who := range actionMessage.Options.To {
						switch who.(type) {
						case string:
							actionMessage.Options.ToGroups = append(actionMessage.Options.ToGroups, gconv.String(who))
						case json.Number:
							uid := gconv.Int(who)
							if uid != 0 {
								actionMessage.Options.ToUsers = append(actionMessage.Options.ToUsers, uid)
							}
						}
					}
				}
				if g.IsEmpty(actionMessage.Options.From) {
					actionMessage.Options.From = es.UserId
				}
			}

			callbackOnMessage(es, actionMessage) // ☆ 消息接收事件
		} else {
			message := "invalid action id"
			if err != nil {
				message = "invalid JSON format"
			}

			es.Write(&JsonObject{
				Status:  http.StatusBadRequest,
				Message: message,
				Data:    gstr.Replace(gconv.String(msg), "\\n", ""),
			})
		}
	}
}

func Broadcast(groupId string, data *JsonObject, excludeFDs ...int64) {
	group.SendMessage(messageFromLocal, &clusterSendMessageObject{
		Data:    data,
		GroupId: groupId,
		Users:   nil,
	}, excludeFDs...)
}
func SendToUsers(groupId string, users []int, data *JsonObject) {
	group.SendMessage(messageFromLocal, &clusterSendMessageObject{
		Data:    data,
		GroupId: groupId,
		Users:   users,
	})
}

func GetWebsocket(groupId string, fd int64) *Websocket {
	return group.GetOne(groupId, fd)
}
func CloseWebsocket(groupId string, fd int64) {
	group.CloseWebsocket(messageFromLocal, groupId, fd)
}
func GroupExists(groupId string) bool {
	return group.GroupExists(messageFromLocal, groupId)
}
func GetUsers(groupId string) []int {
	return group.GetUsers(groupId)
}
func UserExists(groupId string, userId int) bool {
	return group.UserExists(messageFromLocal, groupId, userId)
}

func keepAlive(es *Websocket) {
	lastHeartbeat := gtime.Now()
	// 心跳：响应客户端的pong
	es.websocket.Conn.SetPongHandler(func(msg string) error {
		lastHeartbeat = gtime.Now()
		// fmt.Println(lastHeartbeat.String(), es.FD, "received pong")
		return nil
	})

	// 心跳：收到客户端的 ping
	es.websocket.Conn.SetPingHandler(func(msg string) error {
		lastHeartbeat = gtime.Now()
		// fmt.Println(lastHeartbeat.String(), es.FD, "received ping")
		err := es.websocket.Conn.WriteControl(websocket.PongMessage, pingPongBytes, time.Now().Add(time.Second*5))
		return err
	})

	// 心跳：收到客户端的 p 文本消息
	es.onReceivedPing(func(echo bool) {
		lastHeartbeat = gtime.Now()
		// fmt.Println(lastHeartbeat.String(), es.FD, "received p")
		if echo {
			es.writeText(pingPongBytes)
		}
	})

	if heartbeatTimeout == 0 {
		return
	}

	if heartbeatTimeout < 3 || heartbeatTimeout > 25 {
		heartbeatTimeout = 25
	}

	timeout := time.Duration(heartbeatTimeout) * time.Second

	go func() {
		for {
			time.Sleep(timeout)
			if es.closed {
				return
			}

			// 连续2个周期，如果没有收到客户端的心跳，那么服务端主动发起一个进行探测
			if gtime.Now().Sub(lastHeartbeat) >= timeout*1 {
				// fmt.Println(gtime.Now().String(), es.FD, "did not receive heartbeat, backend will send it.")
				_ = es.websocket.WriteControl(websocket.PingMessage, pingPongBytes, time.Now().Add(time.Second*5))
			}

			// 连续3个周期无心跳，则断开
			if gtime.Now().Sub(lastHeartbeat) > timeout*3 {
				// fmt.Println(gtime.Now().String(), es.FD, "did not receive heartbeat, backend will close it.")
				es.close()
				break
			}
		}
	}()
}
